package com.leiyuee.flink.batch.job;

import com.leiyuee.flink.batch.util.DataFormatUtils;
import com.leiyuee.flink.batch.util.FileUtil;
import com.leiyuee.flink.tools.constants.MySQLConstant;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.data.RowData;
import com.leiyuee.flink.tools.SinkBuilder;


import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;

/**
 * 读取 Txt文件内容，导入Doris,无重复字段
 * 导数时间为下一天上午5：00开始
 *
 * @author Yu_Lei
 * @date 2024/6/27
 */
public class StockJob {
    public static void main(String[] args) throws Exception {
//        log.info("StockJob start");
        LocalDate currentDate = LocalDate.now();
        LocalDate yesterday = currentDate.minusDays(1);
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
        String formattedDate = yesterday.format(formatter);
        //获取下一天日期
        String filePath = "/usr/local/app/CNHUAM0AWSFGW01/s3000-475137724643-hua-icdata/System/GoodsInfo/SAP/Stock/" + formattedDate;
        List<File> filesByDate = FileUtil.getFilesByDate(filePath);
        if (filesByDate.isEmpty()) {
            throw new RuntimeException("导入失败，未读取到文件，请核查：" + formattedDate);
        }
        List<RowData> all = new ArrayList<RowData>();
        try {
            for (File file : filesByDate) {
                System.out.println("读取文件：" + file.getName());
                FileInputStream fileInputStream = new FileInputStream(file);
                InputStreamReader reader = new InputStreamReader(fileInputStream, "UTF-8");
                // 解析文件内容
                BufferedReader bufferedReader = new BufferedReader(reader);
                List<RowData> rowData = DataFormatUtils.praseTxtToRowDataBySort(bufferedReader, 14);
                all.addAll(rowData);
                reader.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("导入失败，数据读取异常！");
        }
        if (all.isEmpty()) {
            throw new RuntimeException("导入失败，未读取到数据");
        }
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        DataStreamSource<RowData> dataStream = env.fromCollection(all);
        String mysqlDorisDB = "Trace_SAPPO";
        String mysqlDorisTB = "tb_sap_stock";
        String mysqlDorisURL = MySQLConstant.DORIS_PRD02_URL + mysqlDorisDB;
        String mysqlDorisStr = String.format("INSERT INTO %s.%s ", mysqlDorisDB, mysqlDorisTB) +
                "(material, plant, sloc, batch, material_description,s, type, storagebin, special_stock_number, bun, available_stock, gr_date)\n" +
                "VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);";
        String indexDorisField = String.format("[%s]%s.%s %s", "Doris MySQL", mysqlDorisDB, mysqlDorisTB, "site");
        SimpleDateFormat dateFormat = new SimpleDateFormat("mm/dd/yyyy");
        DataStreamSink<RowData> rowDataDataStreamSink = dataStream.addSink(SinkBuilder.JdbcSink(mysqlDorisStr,
                (statement, event) -> {
                    statement.setString(1, String.valueOf(event.getString(0)));
                    statement.setString(2, String.valueOf(event.getString(1)));
                    statement.setString(3, String.valueOf(event.getString(2)));
                    statement.setString(4, String.valueOf(event.getString(3)));
                    statement.setString(5, String.valueOf(event.getString(4)));
                    statement.setString(6, String.valueOf(event.getString(5)));
                    statement.setString(7, String.valueOf(event.getString(6)));
                    statement.setString(8, String.valueOf(event.getString(7)));
                    statement.setString(9, String.valueOf(event.getString(9)));
                    statement.setString(10, String.valueOf(event.getString(10)));
                    statement.setString(11, String.valueOf(event.getString(11)));
                    statement.setString(12, String.valueOf(event.getString(12)));
                }
                , mysqlDorisURL, MySQLConstant.DORIS_PRD02_USER_JABIL, MySQLConstant.DORIS_PRD02_USER_JABIL_PWD));

        JobExecutionResult execute = env.execute("StockJob：" + formattedDate + " 数据量：" + all.size());

    }
}
